package cn.spream.jstudy.activemq.server;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.jms.core.JmsTemplate;
import cn.spream.jstudy.activemq.MQType;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * MQ消息接收
 * Created with IntelliJ IDEA.
 * User: sjx
 * Date: 13-6-2
 * Time: 下午6:37
 * To change this template use File | Settings | File Templates.
 */
public class MQReceiver {

    private static final Log log = LogFactory.getLog(MQReceiver.class);

    private JmsTemplate jmsTemplate;
    private static ExecutorService listenerThreads = null;
    private static int threadNumber = 10;

    public void init() {
        log.info("开始-MQ消息处理线程初始化，线程数:" + threadNumber);
        listenerThreads = Executors.newFixedThreadPool(threadNumber);
        log.info("结束-MQ消息处理线程初始化，线程数:" + threadNumber);
    }

    public void destroy(){
        log.info("开始-MQ消息处理线程销毁");
        listenerThreads.shutdown();
        log.info("结束-MQ消息处理线程销毁");
    }

    /**
     * 开启MQ消息监听
     */
    public void listener() {
        log.info("开始-MQ消息处理线程执行MQ监听");
        listenerThreads.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Map<String, Object> message = (Map<String, Object>) jmsTemplate.receiveAndConvert();
                    log.info("收到MQ消息");
                    //收到消息后重新开启一个线程
                    listener();
                    //处理消息
                    MQType mqType = (MQType) message.get("type");
                    Map<String, Object> params = (Map<String, Object>) message.get("params");
                    Processor processor = ProcessorFactory.getProcessor(mqType);
                    execute(processor, params);
                } catch (Exception e) {
                    log.error("MQ消息处理线程出错了，", e);
                }
            }
        });
    }

    /**
     * 根据Type类型进行处理
     */
    private static void execute(Processor processor, Map<String, Object> params) {
        processor.process(params);
    }

    public JmsTemplate getJmsTemplate() {
        return jmsTemplate;
    }

    public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }
}
